package org.zhadoop.streaming.expression;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.common.StreamClassUtil;
import org.zhadoop.streaming.event.IEvent;
import org.zhadoop.streaming.exception.ErrorCode;
import org.zhadoop.streaming.exception.IllegalDataTypeException;
import org.zhadoop.streaming.exception.StreamingException;
import org.zhadoop.streaming.exception.StreamingRuntimeException;
import org.zhadoop.streaming.expression.arithmetic.ICompute;

/**
 * 算术表达式
 * 
 * @author zwd
 *
 */
public class ArithmeticExpression extends OperatorBasedExpression {

	/**
	 * 
	 */
	private static final long serialVersionUID = -3378683666541604538L;

	private static final Logger LOG = LoggerFactory.getLogger(ArithmeticExpression.class);

	private ICompute computer = null;

	public ArithmeticExpression(ExpressionOperator op, IExpression leftExpr, IExpression rightExpr) 
			throws StreamingException{
		super(op, leftExpr, rightExpr);

		Class<?> leftType = StreamClassUtil.getWrapType(leftExpr.getType());
		Class<?> rightType = StreamClassUtil.getWrapType(rightExpr.getType());

		if (!StreamClassUtil.isNumberic(leftType)) {
			StreamingException exception = new StreamingException(
					ErrorCode.SEMANTICANALYZE_ARITHMETIC_EXPRESSION_NUMBER_TYPE);
			LOG.error(ErrorCode.SEMANTICANALYZE_ARITHMETIC_EXPRESSION_NUMBER_TYPE.getFullMessage(), exception);
			throw exception;
		}

		if (!StreamClassUtil.isNumberic(rightType)) {
			StreamingException exception = new StreamingException(
					ErrorCode.SEMANTICANALYZE_ARITHMETIC_EXPRESSION_NUMBER_TYPE);
			LOG.error(ErrorCode.SEMANTICANALYZE_ARITHMETIC_EXPRESSION_NUMBER_TYPE.getFullMessage(), exception);
			throw exception;
		}
		
		setType(validateType());
	}
	
	/**
	 * 根据表达式和运算，预测结果类型
	 * @return
	 */
	public Class<?> validateType() {
		Class<?> resultType = null;
		Class<?> leftType = getLeftExpr().getType();
		Class<?> rightType = getRightExpr().getType();
		
		try {
			if(leftType.equals(rightType)){
				resultType = StreamClassUtil.getWrapType(rightType);
			}else{
				resultType = StreamClassUtil.getArithmaticType(leftType, rightType);
			}
		} catch (IllegalDataTypeException e) {
			throw new IllegalArgumentException(e.getMessage());
		}
		return resultType;
	}

	@Override
	public Object evaluate(IEvent theEvent) {
		if(null == theEvent){
			throw new StreamingRuntimeException("IEvent is null");
		}
		IExpression leftExpr = getLeftExpr();
		IExpression rightExpr = getRightExpr();
		
		Object lo = leftExpr.evaluate(theEvent);
		Object ro = rightExpr.evaluate(theEvent);
		
		return compute(lo,ro);
	}

	/**
	 * 算术运算
	 * @param lo
	 * @param ro
	 * @return
	 */
	private Object compute(Object left, Object right) {
		//如果算术运算的任何一方对象为NULL，则返回结果为NULL
		if(null == left || null == right){
			return null;
		}
		switch(this.getOp()){
			case ADD:
					return computer.add((Number)left,(Number)right);
			case SUBTRACT:
					return computer.add((Number)left, (Number)right);
			case MULTIPLY:
					return computer.multiply((Number)left, (Number)right);
			case DIVIDE:
					return computer.divide((Number)left, (Number)right);
			case MOD:
					return computer.mod((Number)left,(Number)right);
			default:
	                throw new StreamingRuntimeException("unknown relation operator!\n");
		}
	}

	@Override
	public Object evaluate(IEvent[] eventsPerStream) {
		if(null == eventsPerStream || 0 == eventsPerStream.length){
			LOG.error("Streams events are null.");
			throw new StreamingRuntimeException("Streams events are null.");
		}
		
		IExpression leftExpr = getLeftExpr();
		IExpression rightExpr = getRightExpr();
		
		Object lo = leftExpr.evaluate(eventsPerStream);
		Object ro = rightExpr.evaluate(eventsPerStream);
		
		return compute(lo, ro);
	}

}
